package com.shujia.flink.core;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class Demo02BatchWordCount {
    public static void main(String[] args) throws Exception {

        // 1、构建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 设置Flink程序的处理方式：默认是流处理
        /*
         * BATCH：批处理，只能处理有界流，底层是MR模型，可以进行预聚合
         * STREAMING：流处理，可以处理无界流，也可以处理有界流，底层是持续流模型，数据一条一条处理
         * AUTOMATIC：自动判断，当所有的Source都是有界流则使用BATCH模式，当Source中有一个是无界流则会使用STREAMING模式
         */
        env.setRuntimeMode(RuntimeExecutionMode.BATCH);

        // 2、获得第一个DS
        // 通过readTextFile可以基于文件构建有界流
        DataStream<String> wordsFileDS = env.readTextFile("flink/data/words.txt");

        // 3、DS之间的转换
        // 统计每个单词的数量
        // 第一步：将每行数据的每个单词切出来并进行扁平化处理

        // Flink处理逻辑传入的方式
        // new XXXFunction 使用匿名内部类
//        DataStream<String> wordsDS = wordsFileDS.flatMap(new FlatMapFunction<String, String>() {
//            /**
//             *
//             * @param line DS中的一条数据
//             * @param out 通过collect方法将数据发送到下游
//             * @throws Exception
//             */
//            @Override
//            public void flatMap(String line, Collector<String> out) throws Exception {
//                for (String word : line.split(",")) {
//                    // 将每个单词发送到下游
//                    out.collect(word);
//                }
//            }
//        });

        // 使用Lambda表达式
        /*
         * ()->{}
         *  通过 -> 分隔，左边是函数的参数，右边是函数实现的具体逻辑
         */
        DataStream<String> wordsDS = wordsFileDS.flatMap((line, out) -> {
            for (String word : line.split(",")) {
                out.collect(word);
            }
        }, Types.STRING);

        // 使用自定类实现接口中抽象的方法
//        wordsFileDS.flatMap(new MyFunction()).print();

        // 第二步：将每个单词变成 KV格式，V置为1
//        DataStream<Tuple2<String, Integer>> wordKVDS = wordsDS.map(new MapFunction<String, Tuple2<String, Integer>>() {
//            @Override
//            public Tuple2<String, Integer> map(String word) throws Exception {
//                return Tuple2.of(word, 1);
//            }
//        });

        DataStream<Tuple2<String, Integer>> wordKVDS = wordsDS.map(word -> Tuple2.of(word, 1), Types.TUPLE(Types.STRING, Types.INT));


        // 第三步：按每一个单词进行分组
        // keyBy之后数据流会进行分组，相同的key会进入同一个线程中被处理
        // 传递数据的规则：hash取余（线程总数，默认CPU的总线程数）原理
//        KeyedStream<Tuple2<String, Integer>, String> keyedDS = wordKVDS.keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
//            @Override
//            public String getKey(Tuple2<String, Integer> tuple2) throws Exception {
//                return tuple2.f0;
//            }
//        });

        KeyedStream<Tuple2<String, Integer>, String> keyedDS = wordKVDS.keyBy(kv -> kv.f0, Types.STRING);
        // 第四步：对1进行聚合sum
        DataStream<Tuple2<String, Integer>> wordCntDS = keyedDS.sum(1);

        // 4、最终结果的处理（保存/输出打印）
        wordCntDS.print();

        env.execute();

    }
}

class MyFunction implements FlatMapFunction<String,String>{

    @Override
    public void flatMap(String line, Collector<String> out) throws Exception {
        for (String word : line.split(",")) {
                    // 将每个单词发送到下游
                    out.collect(word);
                }
    }
}
